package com.eshipenet.shipnet.dp_decoder

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.scala._
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.json4s.JsonAST.{JObject, JValue}
import org.json4s.native.JsonMethods
import org.json4s.{DefaultFormats, Formats}

object KafkaMessageDeserializer {
  implicit lazy val formats: Formats = DefaultFormats
  val pFields                        = List("encoder", "encoderParams", "data", "id")
}

class KafkaMessageDeserializer extends KafkaDeserializationSchema[EdgeChannelDataPacketUncompressedKafkaMessage] {
  import KafkaMessageDeserializer._

  override def deserialize(
      record: ConsumerRecord[Array[Byte], Array[Byte]]): EdgeChannelDataPacketUncompressedKafkaMessage = {
    val json    = JsonMethods.parse(new String(record.value()))
    val id      = (json \ "id").extract[String]
    val encoder = (json \ "encoder").extract[String]
    val params  = (json \ "encoderParams").extract[Option[JValue]]
    val aux     = JObject(json.filterField(f => !pFields.contains(f._1)))
    val data    = (json \ "data").extract[String].getBytes
    EdgeChannelDataPacketUncompressedKafkaMessage(id, data, encoder, params, aux)
  }

  override def getProducedType: TypeInformation[EdgeChannelDataPacketUncompressedKafkaMessage] =
    createTypeInformation[EdgeChannelDataPacketUncompressedKafkaMessage]

  override def isEndOfStream(nextElement: EdgeChannelDataPacketUncompressedKafkaMessage): Boolean =
    false

}
